﻿using Microsoft.Extensions.Logging;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Collections.Concurrent;
using System.Text;
using XYS.Core.Models;
using XYS.Core.Util;
using XYS.Service.Core.Util.Helper;


namespace XYS.Service.Core.Consumers.Base
{
    /// <summary>
    /// 消费者 基类
    /// </summary>
    public class ConsumerBase<T, T2> : IConsumerBase
        where T : class
    {
        /// <summary>
        /// 日志对象
        /// </summary>
        ILogger<T2> logger;
        private Nest.IElasticClient esClient;

        private string TaskName = "";

        /// <summary>
        /// 
        /// </summary>
        /// <param name="logger"></param>
        public ConsumerBase(ILogger<T2> logger)
        {
            esClient = ElasticClientHelper.GetClient(GlobalCoreSetting.MQSetting.ElasticName);
            this.logger = logger;
            this.TaskName = $"task_{QuqueName}_{Guid.NewGuid()}";
            Task.Run(() =>
            {
                while (true)
                {
                    try
                    {
                        TaskHelper.CreateOrCheckLoopTask(TaskName, DataMap.Count, async (index) =>
                        {
                            try
                            {
                                if (!DataMap.ContainsKey(index) || DataMap[index].IsEmpty) return;

                                ConcurrentDictionary<ulong, T> data = new();
                                foreach (var item in DataMap[index]) data[item.Key] = item.Value;

                                //是否需要记录日志
                                if (this.esClient != null && GlobalCoreSetting.MQSetting.QueueNameList.Contains(QuqueName))
                                {
                                    try
                                    {
                                        //保存记录至ES
                                        string indexName = $"mq-log-{GlobalCoreSetting.ServiceName}-{GlobalCoreSetting.EnvironmentName}-{DateTime.Now:yyyy-MM-dd}".ToLower();
                                        var resultBulkES = esClient.Bulk(indexName, new
                                        {
                                            QueueName = QuqueName,
                                            PrefetchCount = PrefetchCount,
                                            Index = index,
                                            Data = data.Values.ToList(),
                                            CreateTime = DateTime.Now
                                        });
                                        if (!resultBulkES.Succeeded) logger.LogError($"【MQ消费者-保存日志】发生异常.{resultBulkES.ToJson()}");
                                    }
                                    catch (Exception ex) { logger.LogError($"【MQ消费者-保存日志】发生异常.{ex.ToJson()}"); }
                                }

                                try
                                {
                                    await ConsumerReceived(index, data).ConfigureAwait(false);
                                }
                                catch (Exception ex) { logger.LogError(ex.ToJson()); }

                                foreach (var item in data)
                                {
                                    if (DataMap.ContainsKey(index))
                                        DataMap[index].Remove(item.Key, out _);
                                    Tags[index].Remove(item.Key, out _);
                                }
                            }
                            catch (Exception ex) { logger.LogWarning(ex.ToJson()); }
                            finally { await Task.Delay(TimeSpan.FromSeconds(1)); }
                        });
                    }
                    catch (Exception e) { logger.LogError(e.ToJson()); }
                    finally { Thread.Sleep(TimeSpan.FromSeconds(5)); }
                }
            });
        }

        /// <summary>
        /// 队列名
        /// </summary>
        public virtual string QuqueName => throw new NotImplementedException();
        /// <summary>
        /// 每次消费缓存数
        /// </summary>
        public virtual ushort PrefetchCount => 10;
        /// <summary>
        /// 是否自动ACK
        /// </summary>
        public virtual bool AutoACK => false;
        /// <summary>
        /// 消费者对象
        /// </summary>
        public List<AsyncEventingBasicConsumer> AsyncEventingBasicConsumers { get; set; } = new();
        /// <summary>
        /// 待Ack的tag 集合
        /// </summary>
        public List<ConcurrentDictionary<ulong, string>> Tags => tags;
        private List<ConcurrentDictionary<ulong, string>> tags = new();

        /// <summary>
        /// 并发数量
        /// </summary>
        public virtual int ParallelCount => 1;

        /// <summary>
        /// 创建通道
        /// </summary>
        public void CreateChannel(int count)
        {
            for (int i = 0; i < count; i++)
            {
                var channel = MQHelper.CreateChannel(QuqueName, PrefetchCount);
                var asyncEventingBasicConsumer = new AsyncEventingBasicConsumer(channel);

                AsyncEventingBasicConsumers.Add(asyncEventingBasicConsumer);
                if (Tags.Count < AsyncEventingBasicConsumers.Count)
                    Tags.Add(new());

                asyncEventingBasicConsumer.Received += Consumer_Received;

                string consumerTag = channel.BasicConsume(QuqueName, AutoACK, asyncEventingBasicConsumer);
            }
        }


        /// <summary>
        /// 关闭通道
        /// </summary>
        /// <param name="count">关闭通道的数量</param>
        public void CloseChannel(int count)
        {
            var chanels = MQHelper.GetChannels(QuqueName);
            var forCount = chanels.Count - count;

            for (int i = chanels.Count - 1; i >= forCount; i--)
            {
                var chanel = chanels[i];
                AsyncEventingBasicConsumers[i].Received -= Consumer_Received;
                MQHelper.CloseAndRemoveChannel(QuqueName, chanel);
                if (DataMap.ContainsKey(i)) DataMap.Remove(i, out var values);
                AsyncEventingBasicConsumers.RemoveAt(i);
            }
        }


        /// <summary>
        /// 数据集合
        /// </summary>
        ConcurrentDictionary<int, ConcurrentDictionary<ulong, T>> DataMap = new();

        /// <summary>
        /// 消费
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="event"></param>
        /// <returns></returns>
        /// <exception cref="NotImplementedException"></exception>
        public virtual Task Consumer_Received(object sender, BasicDeliverEventArgs @event)
        {
            if (sender is not AsyncEventingBasicConsumer asyncEventingBasicConsumer) return Task.CompletedTask;
            RabbitMQ.Client.IModel channel = asyncEventingBasicConsumer.Model;
            int index = -1;

            for (int i = 0; i < AsyncEventingBasicConsumers.Count; i++)
            {
                if (AsyncEventingBasicConsumers[i].Model == channel)
                {
                    index = i;
                    break;
                }
            }
            if (index == -1) channel.BasicAck(@event.DeliveryTag, false);

            try
            {
                if (@event.Body.Length == 0) { channel.BasicAck(@event.DeliveryTag, false); }
                var json = Encoding.UTF8.GetString(@event.Body.ToArray());
                if (string.IsNullOrWhiteSpace(json)) { BasicNack(index, @event.DeliveryTag, "内容为空"); return Task.CompletedTask; }

                //序列化
                var request = json.ToObject<T>();
                if (request == null) { BasicNack(index, @event.DeliveryTag, "json反序列化失败"); return Task.CompletedTask; }

                if (!DataMap.ContainsKey(index)) DataMap[index] = new();

                DataMap[index][@event.DeliveryTag] = request;
                Tags[index][@event.DeliveryTag] = json;

            }
            catch (Exception e)
            {
                try
                {
                    var json = Encoding.UTF8.GetString(@event.Body.ToArray());
                    MQHelper.Publish($"{@event.RoutingKey}-error", json.ToObject<object>());
                }
                catch
                {
                    try
                    {
                        var json = Encoding.UTF8.GetString(@event.Body.ToArray());
                        MQHelper.Publish($"{@event.RoutingKey}-error", json);
                    }
                    catch { }
                }
                BasicAck(index, @event.DeliveryTag);
            }
            return Task.CompletedTask;
        }


        /// <summary>
        /// 执行
        /// </summary>
        public virtual Task ConsumerReceived(int index, ConcurrentDictionary<ulong, T> DataMap)
        {
            throw new NotImplementedException();
        }

        /// <summary>
        /// BasicNack
        /// </summary>
        /// <param name="index"></param>
        /// <param name="deliveryTag"></param>
        /// <param name="message"></param>
        public virtual void BasicNack(int index, ulong deliveryTag, string message = "")
        {
            if (AsyncEventingBasicConsumers.Count <= index) return;
            lock (AsyncEventingBasicConsumers[index].Model)
            {
                if (!message.IsNullOrWhiteSpace()) logger?.LogWarning(message);
                try
                { AsyncEventingBasicConsumers[index].Model.BasicNack(deliveryTag, false, true); }
                catch
                { }
            }
        }

        /// <summary>
        /// BasicNack
        /// </summary>
        /// <param name="index"></param>
        /// <param name="deliveryTags"></param>
        /// <param name="message"></param>
        public virtual void BasicNack(int index, List<ulong> deliveryTags, string message = "")
        {
            if (AsyncEventingBasicConsumers.Count <= index) return;
            foreach (var deliveryTag in deliveryTags) BasicNack(index, deliveryTag, message.IsNullOrWhiteSpace() ? "" : $"{deliveryTag}:{message}");
        }


        /// <summary>
        /// ACK
        /// </summary>
        /// <param name="index"></param>
        /// <param name="deliveryTag"></param>
        public virtual void BasicAck(int index, ulong deliveryTag)
        {
            if (AsyncEventingBasicConsumers.Count <= index) return;
            lock (AsyncEventingBasicConsumers[index].Model)
            {
                try
                { AsyncEventingBasicConsumers[index].Model.BasicAck(deliveryTag, false); }
                catch
                { this.ReCreateChanel(QuqueName); }
            }
        }

        /// <summary>
        /// ACK
        /// </summary>
        /// <param name="index"></param>
        /// <param name="deliveryTags"></param>
        public virtual void BasicAck(int index, List<ulong> deliveryTags)
        {
            if (AsyncEventingBasicConsumers.Count <= index) return;
            foreach (var deliveryTag in deliveryTags) BasicAck(index, deliveryTag);
        }


    }
}
